package com.freestyle.nettychat.server;


import com.freestyle.netty.easynetty.server.GeneralNettyServerFactory;
import com.freestyle.netty.easynetty.server.interfaces.IGeneralServer;
import com.freestyle.nettychat.chat.ChatCommand;
import com.freestyle.nettychat.chat.CodeConsts;
import com.freestyle.nettychat.chat.attachment.AttachmentDecoder;
import com.freestyle.nettychat.chat.attachment.AttachmentEncoder;
import com.freestyle.nettychat.chat.config.ChatConfig;
import com.freestyle.nettychat.chat.dto.*;
import com.freestyle.nettychat.chat.service.ChatService;
import com.freestyle.nettychat.session.dto.Session;
import com.freestyle.nettychat.session.service.SessionService;
import com.freestyle.tasks.taskqueue.BlockedTaskQueue;
import com.freestyle.tasks.taskqueue.ThreadPool;
import com.freestyle.tasks.taskqueue.interfaces.TaskQueue;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.sound.midi.SysexMessage;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * Netty 聊天室服务器
 * Created by rocklee on 2022/2/5 13:52
 */
@Component
@Slf4j
public class NettyChatServer implements InitializingBean , DisposableBean {
  @Resource
  private SessionService sessionService;
  @Resource
  private ChatService chatService;
  @Resource
  private ChatConfig chatConfig;
  private IGeneralServer server;
  private final ThreadPool threadPool=new ThreadPool(10,10,1800,4);
  private TaskQueue<ChatMessage> queue=new BlockedTaskQueue<>(1000);
  private final AttributeKey<Session> sessionAttributeKey=AttributeKey.valueOf(Session.class,"sessionKey");
  private final AttributeKey<String> currentRoomKey=AttributeKey.valueOf(String.class,"currentRoom");
  private final AttributeKey<Long> firstThreadKey=AttributeKey.valueOf(Long.class,"firstThreadKey");
  private final ConcurrentHashMap<String,ChannelHandlerContext> onLineList=new ConcurrentHashMap<>();

  private Session getSession(ChannelHandlerContext ctx){
    Attribute<Session> sessionAttribute= ctx.channel().attr(sessionAttributeKey);
    Session session=sessionAttribute.get();
    if (session==null){
      throw new IllegalStateException("无效session,请先登录");
    }
    return session;
  }
  private String getUid(ChannelHandlerContext ctx){
    try{
      return getSession(ctx).getUid();
    }
    catch (Exception e){
      return null;
    }
  }
  private String getRoom(ChannelHandlerContext ctx){
    Attribute<String> attribute=ctx.channel().attr(currentRoomKey);
    String room=attribute.get();
    if (room==null){
      throw new IllegalStateException("当前未设置聊天室");
    }
    return room;
  }
  private void logout(ChannelHandlerContext ctx){
    Session session=getSession(ctx);
    try {
      exitRoom(ctx);
    }
    catch(Exception ignored){}
    Attribute<String> attribute=ctx.channel().attr(currentRoomKey);
    attribute.set(null);
    sessionService.logout(session.getUid());
  }
  private void exitRoom(ChannelHandlerContext ctx){
    Session session=getSession(ctx);
    String room=getRoom(ctx);
    sessionService.removeFromChatRoom(session,room);

  }
  private void procCommand(ChannelHandlerContext ctx,CommandMessage command){
    long id=command.getProperties().getId();

    System.out.println(getUid(ctx)+":"+command.getData());
    String[] args= command.getData().split(" ");
    try {
      switch (Objects.requireNonNull(ChatCommand.fromValue(args[0]))) {
        case createRoom:
          sessionService.createChatRoom(args[1], getSession(ctx).getUid());
          break;
        case removeChatRoom:
          if (sessionService.getOnLineChatRoom().containsKey(args[1])) {
            sessionService.removeChatRoom(args[1]);
          }
          else{
            throw new IllegalStateException("无效聊天室:"+args[1]);
          }
          break;
        case login:
          String[] items=args[1].split("\n");
          Session session= sessionService.login(items[0],items[1]);
          session.setConnectionInfo(ctx.channel().remoteAddress().toString());
          Attribute<Session> sessionAttribute= ctx.channel().attr(sessionAttributeKey);
          sessionAttribute.set(session);
          onLineList.put(session.getUid(),ctx);
          break;
        case enterRoom:
          session=getSession(ctx);
          sessionService.grantChatRoom(session.getUid(),args[1]);
          Attribute<String> attribute=ctx.channel().attr(currentRoomKey);
          attribute.set(args[1]);
          break;
        case getOnLineUserList:
          ctx.channel().writeAndFlush(JSONData.fromResult(id,sessionService.getOnLineUserList(args.length==1?null:args[1]).stream()
                  .map(Session::getUid).collect(Collectors.toList())));
          break;
        case getOnLineChatRoom:
          ctx.channel().writeAndFlush(JSONData.fromResult(id,new ArrayList<>(sessionService.getOnLineChatRoom().keySet())));
          break;
        case exit:
          exitRoom(ctx);
          break;
        case logout:
          String uid=getUid(ctx);
          if (uid!=null&&onLineList.containsKey(uid)){
            onLineList.remove(uid);
          }
          logout(ctx);
          break;
        case fetchFile:
          AttachmentMessage attachmentMessage=chatService.recvAttachment(args[1]);
          //ctx.channel().writeAndFlush(LogMessage.fromMessage(command.getProperties().getId(),attachmentMessage));
          ctx.channel().writeAndFlush(attachmentMessage);
          break;
      }
      ctx.channel().writeAndFlush(JSONData.fromResult(id,"操作成功"));
    }
    catch (Exception e){
      ctx.channel().writeAndFlush(JSONData.fromErr(id,1,e.getMessage()));
    }
  }
  private void broadcastOnLineMessage(ChatMessage message){
    List<String> userList=sessionService.getOnLineUserList(message.getProperties().getChatRoom()).stream()
            .map(s->s.getUid()).collect(Collectors.toList());
    for (String uid:userList){
      if (onLineList.containsKey(uid)){
        onLineList.get(uid).channel().writeAndFlush(LogMessage.fromMessage(0,message));
      }
    }
  }
  private IGeneralServer runServer(){
    IGeneralServer server= new GeneralNettyServerFactory().getGeneralServer(chatConfig.getPort());
    try{
      server.getServerBootstrap().option(ChannelOption.SO_BACKLOG, 128) // tcp最大缓存链接个数
              .childOption(ChannelOption.SO_KEEPALIVE, true);
      server.run(ch -> {
        ChannelPipeline pipeline = ch.pipeline();
        // 添加用于处理粘包和拆包问题的处理器
        pipeline.addLast(new LengthFieldBasedFrameDecoder(1024*1024*20, 0, 4, 0, 4))
                .addLast(new LengthFieldPrepender(4))
                .addLast(CodeConsts.createEncoders())
                .addLast(new AttachmentEncoder(CodeConsts.int2Bytes(102)))
                .addLast(new AttachmentDecoder(CodeConsts.int2Bytes(102)))
                .addLast(CodeConsts.createMultiDecoder())
                .addLast(new SimpleChannelInboundHandler(false) {

                  @Override
                  public void channelActive(ChannelHandlerContext ctx) throws Exception {
                    super.channelActive(ctx);
                    System.out.println("Connected:"+ctx.channel().remoteAddress());
                  }
                  /*@Override
                  public boolean acceptInboundMessage(Object msg){
                    //不接受，扔给下一个处理器
                    return false;
                  }*/
                  @Override
                  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                    try {
                      logout(ctx);
                    }
                    catch (Exception ignored){}
                    super.channelInactive(ctx);
                    System.out.println("DisConnected:"+ctx.channel().remoteAddress());
                  }
                  @Override
                  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws
                          Exception {
                    cause.printStackTrace();
                    ctx.close();
                  }
                  @Override
                  protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
                    try {
                      Attribute<Long> tKey=ctx.channel().attr(firstThreadKey);
                      if (tKey.get()==null){
                        tKey.set(Thread.currentThread().getId());
                      }
                      else {
                        if (tKey.get().longValue()!=Thread.currentThread().getId()){
                          System.out.println(String.format("XXXXXXXXXXXXXXXXX不相等，最初为%d,现在为%d",tKey.get(),Thread.currentThread().getId()));
                        }
                      }
                      System.out.println("channelRead0, thread id:"+Thread.currentThread().getId());
                      if (msg instanceof ChatMessage) {
                        ChatMessageProperties properties=((ChatMessage)msg).getProperties();
                        properties.setCtx(ctx);
                        if (!queue.offer((ChatMessage) msg,100)){
                          ctx.channel().writeAndFlush(JSONData.fromErr(0,-1,"任务队列满，系统正忙请稍后再试"));
                          return;
                        }
                      }
                    }
                    catch (Exception e){
                      e.printStackTrace();
                      ctx.channel().writeAndFlush(JSONData.fromErr(0,-1,e.getMessage()));
                    }
                  }
                });
      },false);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    finally {

    }
    return server;
  }
  private void initThreads(Consumer<ChatMessage> consumer){
    List<CompletableFuture> list=new ArrayList<>();
    for (int i=0;i<threadPool.getCorePoolSize();i++){
      list.add(CompletableFuture.runAsync(()->{
        while (!closed.get()){
          ChatMessage message=queue.poll(3000);
          if (message==null){
            continue;
          }
          consumer.accept(message);
        }
      }));
    }
    tasks=list.toArray(new CompletableFuture[0]);
  }
  public void await(){
    server.await();
  }
  private AtomicBoolean closed=new AtomicBoolean();
  public void close(){
    server.close();
    queue.clear();
    queue.close();
    closed.set(true);
    CompletableFuture.allOf(tasks).join();
    log.info("netty server is closed.");
  }
  private CompletableFuture[] tasks;

  @Override
  public void afterPropertiesSet() throws Exception {
    initThreads(chatMessage->{
      ChannelHandlerContext ctx= (ChannelHandlerContext) chatMessage.getProperties().getCtx();
      try {
        CodeConsts.MessageType messageType = CodeConsts.MessageType.fromValue(chatMessage.getProperties().getSClass());
        switch (Objects.requireNonNull(messageType)) {
          case text:
          case attachment:
            getSession(ctx);
            chatService.sendMessage(chatMessage);
            broadcastOnLineMessage(chatMessage);
            break;
          case command:
            procCommand(ctx, (CommandMessage) chatMessage);
            break;
        }
        System.out.println("当前线程ID:"+Thread.currentThread().getId()+"，当前任务队列数：" + queue.count());
      }
      catch (Exception e){
        e.printStackTrace();
        ctx.channel().writeAndFlush(JSONData.fromErr(0,-1,e.getMessage()));
      }
    });
   server= runServer();
  }

  @Override
  public void destroy() throws Exception {
    close();
  }
}
